package org.example.realtime.gongdan.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class UserTagsJob {
    public static void main(String[] args) {
        TableEnvironment tableEnv = getTableEnv();

        createInputTable(tableEnv);

        handle(tableEnv);

        createOutputTable(tableEnv);

        saveStarrocks(tableEnv);

    }

    private static void saveStarrocks(TableEnvironment tableEnv) {
//        tableEnv.executeSql("SELECT * FROM user_tags_table").print();
        tableEnv.executeSql(
                "INSERT INTO `dws_user_tags` " +
                        "select * " +
                        "from user_tags_table"
        );
    }

    private static void handle(TableEnvironment tableEnv) {

        // 执行ETL逻辑
        tableEnv.executeSql(
                "CREATE TEMPORARY FUNCTION get_json_string AS 'org.example.realtime.gongdan.log.job.JsonExtractorUDF'"
        );

//        Table table = tableEnv.sqlQuery(
//                "SELECT " +
//                        "  r.user_id, " +
//                        "  r.province_id, " +
//                        "  p.province_name AS residential_province, " +
//                        "  r.city_id, " +
//                        "  p.city_name AS residential_city, " +
//                        "  p.city_grade, " +
//                        "  CASE " +
//                        "    WHEN p.province_name IN ('辽宁省', '吉林省', '黑龙江省') THEN '东北' " +
//                        "    WHEN p.province_name IN ('上海省', '江苏省', '浙江省', '安徽省', '福建省', '江西省', '山东省') THEN '华东' " +
//                        "    WHEN p.province_name IN ('湖北省', '湖南省', '河南省') THEN '华中' " +
//                        "    WHEN p.province_name IN ('北京市', '天津市', '河北省', '山西省', '内蒙古自治区') THEN '华北' " +
//                        "    WHEN p.province_name IN ('广东省', '广西壮族自治区', '海南省') THEN '华南' " +
//                        "    WHEN p.province_name IN ('香港省', '澳门省', '台湾省') THEN '港澳台' " +
//                        "    WHEN p.province_name IN ('陕西省', '甘肃省', '宁夏回族自治区', '青海省', '新疆维吾尔自治区') THEN '西北' " +
//                        "    WHEN p.province_name IN ('重庆省', '四川省', '贵州省', '云南省', '西藏自治区') THEN '西南' " +
//                        "    ELSE '其他' END AS resident_region, " +
//                        "  CONCAT_WS(',', " +
//                        "    get_json_string(w.living_index, '$[0].name'), " +
//                        "    get_json_string(w.living_index, '$[0].level'), " +
//                        "    get_json_string(w.living_index, '$[0].text') " +
//                        "  ) AS clothing_index, " +
//                        "  CONCAT_WS(',', " +
//                        "    get_json_string(w.living_index, '$[1].name'), " +
//                        "    get_json_string(w.living_index, '$[1].level'), " +
//                        "    get_json_string(w.living_index, '$[1].text') " +
//                        "  ) AS car_wash_index, " +
//                        "  CONCAT_WS(',', " +
//                        "    get_json_string(w.living_index, '$[2].name'), " +
//                        "    get_json_string(w.living_index, '$[2].level'), " +
//                        "    get_json_string(w.living_index, '$[2].text') " +
//                        "  ) AS sport_index, " +
//                        "  get_json_string(w.lives, '$.humidity') AS humidity, " +
//                        "  get_json_string(w.lives, '$.temperature') AS temperature, " +
//                        "  get_json_string(w.lives, '$.weather') AS weather, " +
//                        "  get_json_string(w.lives, '$.winddirection') AS winddirection, " +
//                        "  get_json_string(w.lives, '$.windpower') AS windpower, " +
//                        "  DATE_FORMAT(w.report_time, 'yyyy-MM-dd HH:mm:ss') AS update_date " +
//                        "FROM (\n" +
//                        "         SELECT\n" +
//                        "             user_id,\n" +
//                        "             province_id,\n" +
//                        "             city_id,\n" +
//                        "             address_count,\n" +
//                        "             last_used_time\n" +
//                        "         FROM (\n" +
//                        "                  SELECT\n" +
//                        "                      user_id,\n" +
//                        "                      province_id,\n" +
//                        "                      city_id,\n" +
//                        "                      COUNT(*) AS address_count,\n" +
//                        "                      MAX(order_time) AS last_used_time\n" +
//                        "                  FROM dwd_user_orders\n" +
//                        "                  WHERE order_time >= LOCALTIMESTAMP - INTERVAL '180' DAY(3)\n" +
//                        "                  GROUP BY user_id, province_id, city_id\n" +
//                        "              ) t\n" +
//                        "     )r " +
//                        "LEFT JOIN dim_province_table p " +
//                        "  ON r.province_id = p.province_id AND r.city_id = p.city_id " +
//                        "LEFT JOIN dim_spider_amap_weather_data_dtl w " +
//                        "  ON r.city_id = w.code "
//        );


        Table table = tableEnv.sqlQuery(
                "SELECT " +
                "  r.user_id, " +
                "  r.province_id, " +
                "  p.province_name AS residential_province, " +
                "  r.city_id, " +
                "  p.city_name AS residential_city, " +
                "  p.city_grade, " +
                "  CASE " +
                "    WHEN p.province_name IN ('辽宁省', '吉林省', '黑龙江省') THEN '东北' " +
                "    WHEN p.province_name IN ('上海市', '江苏省', '浙江省', '安徽省', '福建省', '江西省', '山东省') THEN '华东' " +
                "    WHEN p.province_name IN ('湖北省', '湖南省', '河南省') THEN '华中' " +
                "    WHEN p.province_name IN ('北京市', '天津市', '河北省', '山西省', '内蒙古自治区') THEN '华北' " +
                "    WHEN p.province_name IN ('广东省', '广西壮族自治区', '海南省') THEN '华南' " +
                "    WHEN p.province_name IN ('香港省', '澳门省', '台湾省') THEN '港澳台' " +
                "    WHEN p.province_name IN ('陕西省', '甘肃省', '宁夏回族自治区', '青海省', '新疆维吾尔自治区') THEN '西北' " +
                "    WHEN p.province_name IN ('重庆市', '四川省', '贵州省', '云南省', '西藏自治区') THEN '西南' " +
                "    ELSE '其他' END AS resident_region, " +
                "  CONCAT_WS(',', " +
                "    get_json_string(w.living_index, '$[0].name'), " +
                "    get_json_string(w.living_index, '$[0].level'), " +
                "    get_json_string(w.living_index, '$[0].text') " +
                "  ) AS clothing_index, " +
                "  CONCAT_WS(',', " +
                "    get_json_string(w.living_index, '$[1].name'), " +
                "    get_json_string(w.living_index, '$[1].level'), " +
                "    get_json_string(w.living_index, '$[1].text') " +
                "  ) AS car_wash_index, " +
                "  CONCAT_WS(',', " +
                "    get_json_string(w.living_index, '$[2].name'), " +
                "    get_json_string(w.living_index, '$[2].level'), " +
                "    get_json_string(w.living_index, '$[2].text') " +
                "  ) AS sport_index, " +
                "  get_json_string(w.lives, '$.humidity') AS humidity, " +
                "  get_json_string(w.lives, '$.temperature') AS temperature, " +
                "  get_json_string(w.lives, '$.weather') AS weather, " +
                "  get_json_string(w.lives, '$.winddirection') AS winddirection, " +
                "  get_json_string(w.lives, '$.windpower') AS windpower, " +
                "  DATE_FORMAT(w.report_time, 'yyyy-MM-dd HH:mm:ss') AS update_date " +
                "FROM (\n" +
                "         SELECT\n" +
                "             user_id,\n" +
                "             province_id,\n" +
                "             city_id,\n" +
                "             address_count,\n" +
                "             last_used_time\n" +
                "         FROM (\n" +
                "                  SELECT\n" +
                "                      user_id,\n" +
                "                      province_id,\n" +
                "                      city_id,\n" +
                "                      COUNT(*) AS address_count,\n" +
                "                      MAX(order_time) AS last_used_time\n" +
                "                  FROM dwd_user_orders\n" +
                "                  WHERE order_time >= LOCALTIMESTAMP - INTERVAL '180' DAY(3)\n" +
                "                  GROUP BY user_id, province_id, city_id\n" +
                "              ) t\n" +
                "     )r " +
                "LEFT JOIN dim_province_table p " +
                "  ON r.province_id = p.province_id AND r.city_id = p.city_id " +
                "LEFT JOIN dim_spider_amap_weather_data_dtl w " +
                "  ON r.city_id = w.code " +
                "UNION ALL " +
                        "SELECT " +
                        "  l.user_id, " +
                        "  p.province_id, " +
                        "  p.province_name AS residential_province, " +
                        "  p.city_id, " +
                        "  p.city_name AS residential_city, " +
                        "  p.city_grade, " +
                        "  CASE " +
                        "    WHEN p.province_name IN ('辽宁省', '吉林省', '黑龙江省') THEN '东北' " +
                        "    WHEN p.province_name IN ('上海市', '江苏省', '浙江省', '安徽省', '福建省', '江西省', '山东省') THEN '华东' " +
                        "    WHEN p.province_name IN ('湖北省', '湖南省', '河南省') THEN '华中' " +
                        "    WHEN p.province_name IN ('北京市', '天津市', '河北省', '山西省', '内蒙古自治区') THEN '华北' " +
                        "    WHEN p.province_name IN ('广东省', '广西壮族自治区', '海南省') THEN '华南' " +
                        "    WHEN p.province_name IN ('香港省', '澳门省', '台湾省') THEN '港澳台' " +
                        "    WHEN p.province_name IN ('陕西省', '甘肃省', '宁夏回族自治区', '青海省', '新疆维吾尔自治区') THEN '西北' " +
                        "    WHEN p.province_name IN ('重庆市', '四川省', '贵州省', '云南省', '西藏自治区') THEN '西南' " +
                        "    ELSE '其他' END AS resident_region, " +
                        "  CONCAT_WS(',', " +
                        "    get_json_string(w.living_index, '$[0].name'), " +
                        "    get_json_string(w.living_index, '$[0].level'), " +
                        "    get_json_string(w.living_index, '$[0].text') " +
                        "  ) AS clothing_index, " +
                        "  CONCAT_WS(',', " +
                        "    get_json_string(w.living_index, '$[1].name'), " +
                        "    get_json_string(w.living_index, '$[1].level'), " +
                        "    get_json_string(w.living_index, '$[1].text') " +
                        "  ) AS car_wash_index, " +
                        "  CONCAT_WS(',', " +
                        "    get_json_string(w.living_index, '$[2].name'), " +
                        "    get_json_string(w.living_index, '$[2].level'), " +
                        "    get_json_string(w.living_index, '$[2].text') " +
                        "  ) AS sport_index, " +
                        "  get_json_string(w.lives, '$.humidity') AS humidity, " +
                        "  get_json_string(w.lives, '$.temperature') AS temperature, " +
                        "  get_json_string(w.lives, '$.weather') AS weather, " +
                        "  get_json_string(w.lives, '$.winddirection') AS winddirection, " +
                        "  get_json_string(w.lives, '$.windpower') AS windpower, " +
                        "  DATE_FORMAT(w.report_time, 'yyyy-MM-dd HH:mm:ss') AS update_date " +
                        "FROM dwd_user_location l " +
                        "LEFT JOIN dim_province_table p " +
                        "  ON (p.province_name IN ('北京市', '上海市', '天津市', '重庆市') AND l.province_name = p.province_name) " +
                        "  OR (p.province_name NOT IN ('北京市', '上海市', '天津市', '重庆市') and l.city_name LIKE CONCAT('%', p.city_name, '%')) " +
                        "LEFT JOIN dim_spider_amap_weather_data_dtl w " +
                        "  ON p.city_name = w.city " +
                        "WHERE NOT EXISTS ( " +
                        "  SELECT 1 FROM dwd_user_orders o " +
                        "  WHERE o.user_id = l.user_id " +
                        "  AND o.order_time >= LOCALTIMESTAMP - INTERVAL '180' DAY(3) " +
                        ") "
        );


        tableEnv.createTemporaryView("user_tags_table",table);

    }

    private static void createOutputTable(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS dws_user_tags (" +
                        "`user_id` BIGINT NOT NULL COMMENT '用户ID'," +
                        "`province_id` BIGINT NOT NULL COMMENT '省份ID'," +
                        "`residential_province` STRING COMMENT '居住省份'," +
                        "`city_id` BIGINT NOT NULL COMMENT '城市ID'," +
                        "`residential_city` STRING COMMENT '居住城市'," +
                        "`city_grade` INT COMMENT '城市等级(1-6)'," +
                        "`resident_region` STRING COMMENT '常驻区域'," +
                        "`clothing_index` STRING COMMENT '穿衣指数'," +
                        "`car_wash_index` STRING COMMENT '洗车指数'," +
                        "`sport_index` STRING COMMENT '运动指数'," +
                        "`humidity` STRING COMMENT '湿度指数'," +
                        "`temperature` STRING COMMENT '温度指数'," +
                        "`weather` STRING COMMENT '天气'," +
                        "`winddirection` STRING COMMENT '风向'," +
                        "`windpower` STRING COMMENT '风力指数'," +
                        "`update_date` STRING COMMENT '标签更新日期'," +
                        "PRIMARY KEY (user_id) NOT ENFORCED" +
                        ") WITH (" +
                        "'connector' = 'starrocks',\n" +
                        "'jdbc-url' = 'jdbc:mysql://node101:9030',\n" +
                        "'load-url' = 'node101:8030',\n" +
                        "'database-name' = 'jtp_data_warehouse',\n" +
                        "'table-name' = 'dws_user_tags',\n" +
                        "'username' = 'root',\n" +
                        "'password' = '123456',\n" +
                        "'sink.properties.format' = 'json',\n" +
                        "'sink.properties.strip_outer_array' = 'true'\n" +
                        ")"
        );
    }

    private static void createInputTable(TableEnvironment tableEnv) {
        // 创建天气数据表
        tableEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS dim_spider_amap_weather_data_dtl (" +
                        "  `code` INT COMMENT '城市代码'," +
                        "  `report_time` TIMESTAMP(3) COMMENT '数据更新时间'," +
                        "  `province` STRING COMMENT '省份'," +
                        "  `city` STRING COMMENT '城市'," +
                        "  `lives` STRING COMMENT '实时天气数据(JSON)'," +
                        "  `living_index` STRING COMMENT '生活指数数据(JSON)'," +
                        "  `update_time` TIMESTAMP(3) COMMENT '数据入库时间'" +
                        ") WITH (" +
                        "  'connector' = 'jdbc'," +
                        "  'url' = 'jdbc:mysql://node101:9030/jtp_data_warehouse'," +
                        "  'table-name' = 'spider_amap_weather_data_dtl'," +
                        "  'lookup.cache.max-rows' = '7'," +
                        "  'lookup.cache.ttl' = '1 hour'," +
                        "  'username' = 'root'," +
                        "  'password' = '123456'," +
                        "  'driver' = 'com.mysql.cj.jdbc.Driver'" +
                        ")"
        );

        // 创建用户订单表
        tableEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS dwd_user_orders (" +
                        "  `order_id` BIGINT COMMENT '订单ID'," +
                        "  `user_id` BIGINT COMMENT '用户ID'," +
                        "  `province_id` BIGINT COMMENT '收货省份ID'," +
                        "  `city_id` BIGINT COMMENT '收货城市ID'," +
                        "  `order_time` TIMESTAMP(3) COMMENT '下单时间'," +
                        "  WATERMARK FOR `order_time` AS `order_time` - INTERVAL '5' SECOND" +
                        ") WITH (" +
                        "  'connector' = 'jdbc'," +
                        "  'url' = 'jdbc:mysql://node101:9030/jtp_data_warehouse'," +
                        "  'table-name' = 'dwd_user_orders'," +
                        "  'username' = 'root'," +
                        "  'password' = '123456'," +
                        "  'driver' = 'com.mysql.cj.jdbc.Driver'" +
                        ")"
        );

        // 创建省份维度表
        tableEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS dim_province_table (" +
                        "  `province_id` INT NOT NULL COMMENT '省份ID'," +
                        "  `city_id` INT NOT NULL COMMENT '城市ID'," +
                        "  `province_name` STRING COMMENT '省份名称'," +
                        "  `city_name` STRING COMMENT '城市名称'," +
                        "  `city_grade` INT COMMENT '城市等级(1-6)'" +
                        ") WITH (" +
                        "  'connector' = 'jdbc'," +
                        "  'url' = 'jdbc:mysql://node101:9030/jtp_data_warehouse'," +
                        "  'table-name' = 'dim_province_table'," +
                        "  'lookup.cache.max-rows' = '5'," +
                        "  'lookup.cache.ttl' = '1 hour'," +
                        "  'username' = 'root'," +
                        "  'password' = '123456'," +
                        "  'driver' = 'com.mysql.cj.jdbc.Driver'" +
                        ")"
        );

        tableEnv.executeSql(
                "CREATE TABLE dwd_user_location (" +
                        "  user_id BIGINT NOT NULL COMMENT '用户ID（Unique Key）'," +
                        "  city_name VARCHAR(50) COMMENT '城市名称'," +
                        "  province_name VARCHAR(50) COMMENT '省份名称'" +
                        ") WITH (" +
                        "  'connector' = 'jdbc'," +
                        "  'url' = 'jdbc:mysql://node101:9030/jtp_data_warehouse'," +
                        "  'table-name' = 'dwd_user_location'," +
                        "  'lookup.cache.max-rows' = '5'," +
                        "  'lookup.cache.ttl' = '1 hour'," +
                        "  'username' = 'root'," +
                        "  'password' = '123456'," +
                        "  'driver' = 'com.mysql.cj.jdbc.Driver'" +
                        ")"
        );

//        tableEnv.executeSql("select * from dim_province_table").print();

    }

    private static TableEnvironment getTableEnv() {
//        EnvironmentSettings build = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
//        TableEnvironment tableEnv = TableEnvironment.create(build);
//
//        Configuration configuration = tableEnv.getConfig().getConfiguration();
//        configuration.setString("table.local-time-zone","Asia/Shanghai");
//        configuration.setString("table.exec.resource.default-parallelism","1");
//        configuration.setString("table.exec.state.ttl","5s");
//        configuration.setString("table.exec.sink.upsert-materialize", "none");
//
//        return tableEnv;



        // 环境属性
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment taEnv = TableEnvironment.create(settings) ;
        // 配置属性
        Configuration configuration = taEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        taEnv.getConfig().getConfiguration().setString(
                "table.exec.sink.not-null-enforcer", "drop");
        // 返回对象
        return taEnv;

    }
}